/*******************************************************************************
 * Copyright (c) 2010 Mikhail Garber.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the GNU Lesser Public License (LGPL)
 * which accompanies this distribution
 * 
 * Contributors:
 *     Mikhail Garber - initial API and implementation
 ******************************************************************************/
package plumber.coordinator;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import javax.sql.DataSource;


import plumber.common.HasSetup;
import plumber.common.Utils;
import plumber.common.WorkItem;
import plumber.common.WorkItem.Status;

/**
 * Coordinator that uses a database table to store list of WorkItems
 * Allows for:
 * distributed execution - Clients and multiple WorkManagers can access one database remotely
 * scalability - you can add more capacity and reduce workflow latency by adding more WorkManagers 
 * (more servers or more JVMs)
 * smooth optimistic locking - this will scale for as long as contention on the WorkItems table is not too bad
 * 
 * @author mgarber
 *
 */
public class SimpleJdbcCoordinator extends AbstractPersistentCoordinator implements
		CoordinatorClient, CoordinatorServer, HasSetup {

	private static final String TABLE_NAME = "plumber_work";

	private DataSource datasource;

	public void setDatasource(DataSource datasource) {
		this.datasource = datasource;
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorServer#getCandidates(java.lang.String, int)
	 */
	@Override
	public List<String> getCandidates(String namespace, int upto) {
		return this.contextConverter.toStorableStrings(makeWorkItems(query("select * from " + TABLE_NAME
				+ " where " + STATUS_COLUMN + " = ? and "
				+ DATE_SCHEDULED_COLUMN + " < ? and " +
				NAMESPACE_COLUMN + " = ? " +
				"order by " + DATE_SCHEDULED_COLUMN, upto,
				WorkItem.Status.AVAILABLE.toString(),
				new Timestamp(System.currentTimeMillis()),
				namespace)));
	}

	private MyWorkItem find(String id) {
		List<MyWorkItem> found = query("select * from " + TABLE_NAME
				+ " where " + PK_COLUMN + " = ?", id);
		if (found.size() == 0) {
			return null;
		}
		if (found.size() > 1) {
			throw new RuntimeException("more than one item matched id");
		}
		return found.get(0);
	}

	@Override
	public boolean updateStatus(String id, String fromStatus, String status, String context) {
		Connection conn = getConnectionFromPool();
		boolean ret = updateStatus(conn, id, fromStatus, status, context);
		commitAndDispose(conn);
		return ret;
	}

	private boolean updateStatus(Connection conn, String id, String fromStatus, String status, String context) {
		MyWorkItem item = find(id);
		if (item == null) {
			return false;
		}
		if(item.getStatus() != Status.valueOf(fromStatus)) {
			return false;
		}
		if(context == null) {
			context = item.getContext();
		}
		int count = update(conn, "update " + TABLE_NAME + " set "
				+ STATUS_COLUMN + " = ?, " + 
				CONTEXT_COLUMN + " = ?, " +
				VERSION_COLUMN + " = ? where " + PK_COLUMN + " = ? and "
				+ VERSION_COLUMN + " = ?", status, context, item.getVersion() + 1, item.getId(),
				item.getVersion());
		return (count == 1);
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorServer#advanceWork(java.lang.String, java.lang.String, java.lang.String, plumber.common.WorkItem.Status, java.util.Date)
	 */
	@Override
	public void advanceWork(String context, String oldItemId,
			String nextStepId, Status oldStatus, Date scheduled) {
		logger.info("advancing work: " + context + " "
				+ oldStatus + " " + nextStepId);
		MyWorkItem oldItem = find(oldItemId);
		Utils.validatePresent(oldItem, "old item");
		if (nextStepId != null) {
			WorkItem newItem = makeNewItem(context, oldItem.getFlowName(), nextStepId, scheduled,
					oldItem.getRequestId(), oldItem.getNamespace());
			Connection conn = getConnectionFromPool();
			if (!updateStatus(conn, oldItemId, Status.IN_PROGRESS.toString(), 
					oldStatus.toString(), context)) {
				dispose(conn, null, null);
				throw new RuntimeException("was not abie to update status");
			}
			create(conn, newItem);
			commitAndDispose(conn);
		} else {
			updateStatus(oldItemId, Status.IN_PROGRESS.toString(),
					oldStatus.toString(), context);
		}
	}

	@Override
	public void create(WorkItem item) {
		Connection conn = getConnectionFromPool();
		create(conn, item);
		commitAndDispose(conn);
	}

	private void create(Connection conn, WorkItem item) {
		logger.info("creating:" + item);
		update(conn, "insert into " + TABLE_NAME + "(" + PK_COLUMN + ","
				+ CONTEXT_COLUMN + "," + DATE_CREATED_COLUMN + ","
				+ DATE_SCHEDULED_COLUMN + "," + RID_COLUMN + ","
				+ STATUS_COLUMN + "," + STEP_ID_COLUMN + "," + VERSION_COLUMN 
				+ ", " + FLOW_NAME_COLUMN 
				+ ", " + NAMESPACE_COLUMN 
				+ ") values (?,?,?,?,?,?,?,?,?,?)", item.getId(), item.getContext()
				, new Timestamp(System.currentTimeMillis()),
				new Timestamp(System.currentTimeMillis()), item.getRequestId(),
				item.getStatus().toString(), item.getFlowStepId(), 1, item.getFlowName(),
				item.getNamespace());
	}

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorClient#getHistory(java.lang.String)
	 */
	@Override
	public List<String> getHistory(String requestId) {
		return this.contextConverter.toStorableStrings(makeWorkItems(query("select * from " + TABLE_NAME
				+ " where " + RID_COLUMN + " = ?", requestId)));
	}

	
	@Override
	public List<MyWorkItem> itemsMatchingRequestIdAndStatus(String requestId,
			String status) {
		List<MyWorkItem> items = query(
				"select * from " + TABLE_NAME + " where " + RID_COLUMN
						+ " = ? and " + STATUS_COLUMN + " = ?", requestId,
				status);
		return items;
	}

	/* (non-Javadoc)
	 * @see plumber.common.HasSetup#setup()
	 */
	@Override
	public void setup() {
		try {
			String sql = "create table " + TABLE_NAME + "(" + PK_COLUMN
			+ " varchar(40) primary key, " + RID_COLUMN
			+ " varchar(40) not null," + DATE_CREATED_COLUMN
			+ " timestamp, " + DATE_SCHEDULED_COLUMN + " timestamp, "
			+ VERSION_COLUMN + " integer, " + CONTEXT_COLUMN
			+ " varchar(2000), " + STEP_ID_COLUMN + " varchar (30), "
			+ FLOW_NAME_COLUMN + " varchar(40), "
			+ NAMESPACE_COLUMN + " varchar(40), "
			+ STATUS_COLUMN + " varchar(20))";
			logger.info(sql);
			update(sql);
		} catch (RuntimeException re) {
			logger.warn("setting up", re);
		}
	}

	@Override
	public void shutdown() {
		// TODO Auto-generated method stub

	}

	private List<MyWorkItem> query(String sql, Object... params) {
		return query(sql, Integer.MAX_VALUE, params);
	}
	
	private List<MyWorkItem> query(String sql, int upto, Object... params) {
		Connection conn = null;
		PreparedStatement stmt = null;
		ResultSet rs = null;
		try {
			conn = datasource.getConnection();
			stmt = conn.prepareStatement(sql);
			for (int i = 1; i <= params.length; i++) {
				stmt.setObject(i, params[i-1]);
			}
			rs = stmt.executeQuery();
			List<MyWorkItem> result = new ArrayList<MyWorkItem>();
			int i=0;
			while (rs.next() && i<upto) {
				result.add(fromRS(rs));
				i++;
			}
			logger.info("result:" + result);
			return result;
		} catch (SQLException e) {
			logger.warn("sql", e);
			throw new RuntimeException(e);
		} finally {
			dispose(conn, stmt, rs);
		}

	}

	private int update(String sql, Object... params) {
		Connection conn = getConnectionFromPool();
		int ret = update(conn, sql, params);
		commitAndDispose(conn);
		return ret;
	}

	private void commitAndDispose(Connection conn) {
		try {
			conn.commit();
			logger.info("commit");
		} catch (SQLException e) {
			throw new RuntimeException(e);
		} finally {
			dispose(conn, null, null);
		}
	}

	private int update(Connection conn, String sql, Object... params) {
		PreparedStatement stmt = null;
		try {
			stmt = conn.prepareStatement(sql);
			
			for (int i = 1; i <= params.length; i++) {
				
				stmt.setObject(i, params[i-1]);
			}
			int ret = stmt.executeUpdate();

			return ret;
		} catch (SQLException e) {
			logger.warn("sql", e);
			throw new RuntimeException(e);
		} finally {
			dispose(null, stmt, null);
		}

	}

	private Connection getConnectionFromPool() {
		Connection conn = null;
		try {
			conn = datasource.getConnection();
			conn.setAutoCommit(false);
		} catch (SQLException e) {
			dispose(conn, null, null);
			throw new RuntimeException(e);
		}
		return conn;
	}

	private void dispose(Connection conn, PreparedStatement stmt, ResultSet rs) {
		if (rs != null) {
			try {
				rs.close();
			} catch (SQLException e) {
				logger.warn("ignoring", e);
			}
		}
		if (stmt != null) {
			try {
				stmt.close();
			} catch (SQLException e) {
				logger.warn("ignoring", e);
			}
		}
		if (conn != null) {
			try {
				conn.close();
			} catch (SQLException e) {
				logger.warn("ignoring", e);
			}
		}
	}

	private MyWorkItem fromRS(ResultSet rs) throws SQLException {
		MyWorkItem w = new MyWorkItem();
		
		w.setContext(rs.getString(CONTEXT_COLUMN));
		w.setFlowStepId(rs.getString(STEP_ID_COLUMN));
		w.setId(rs.getString(PK_COLUMN));
		w.setRequestId(rs.getString(RID_COLUMN));
		w.setScheduled(rs.getTimestamp(DATE_SCHEDULED_COLUMN));
		w.setStatus(Status.valueOf(rs.getString(STATUS_COLUMN)));
		w.setVersion(rs.getInt(VERSION_COLUMN));
		w.setFlowName(rs.getString(FLOW_NAME_COLUMN));
		w.setNamespace(rs.getString(NAMESPACE_COLUMN));
		
		return w;
	}

	

	/* (non-Javadoc)
	 * @see plumber.coordinator.CoordinatorClient#getAll(java.lang.String, int)
	 */
	@Override
	public List<String> getAll(String namespace, int upto) {
		return this.contextConverter.toStorableStrings(makeWorkItems(query("select * from " + TABLE_NAME
				+ " where " + NAMESPACE_COLUMN + " = ? "
				+ " order by " + DATE_SCHEDULED_COLUMN, upto, namespace)));
	}

	public void purge() {
		update("delete from " + TABLE_NAME);
	}
}
